package com.eet.elasticsearch.repository;

import com.alibaba.fastjson.JSON;
import com.eet.elasticsearch.model.UserSearchParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

/**
 * 依靠 ES api提供的 Client对象来完成增删改查功能
 *
 * @author Administrator
 */
@Slf4j
@Component
public class ElasticOperationRepository {

    @Autowired
    private Client client;

    private BulkProcessor bulkProcessor;

    @PostConstruct
    public void initBulkProcessor() {
        bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest bulkRequest) {
                log.info("序号：{} 开始执行{}条记录保存", executionId, bulkRequest.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                log.info("序号：{}执行{}条记录保存成功，耗时：{}毫秒，", executionId, bulkRequest.numberOfActions(), bulkResponse.getTookInMillis());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest bulkRequest, Throwable failure) {
                log.error(String.format("序号：%s执行失败，总记录数：%s", executionId, bulkRequest.numberOfActions()), failure);
            }
        }).setBulkActions(1000)
                .setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB))
                .setConcurrentRequests(4)
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                // 失败后等待多久及重试次数
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueSeconds(500), 3))
                .build();
    }

    @PreDestroy
    public void closeBulk() {
        if (bulkProcessor != null) {
            try {
                bulkProcessor.close();
            } catch (Exception e) {
                log.error("close bulkProcessor exception：{}", e.getMessage());
            }
        }
    }

    /**
     * 批量添加,性能最好
     *
     * @param indices 指标
     * @param type    类型
     * @param object  对象
     */
    public void addDocumentBulkProcess(String indices, String type, Object object) {
        bulkProcessor.add(client.prepareIndex(indices, type).setSource(JSON.toJSONString(object), XContentType.JSON).request());
    }

    /**
     * 添加文档
     *
     * @param indices 指标
     * @param type    类型
     * @param object  对象
     */
    public void addDocument(String indices, String type, Object object) {
        IndexResponse indexResponse = client.prepareIndex(indices, type).setSource(JSON.toJSONString(object), XContentType.JSON).get();
        log.info("添加结果：{}", indexResponse.toString());
    }

    /**
     * 按照ID删除
     *
     * @param index 索引
     * @param type  类型
     * @param id    ID
     */
    public void deleteDocumentById(String index, String type, String id) {
        DeleteResponse response = client.prepareDelete(index, type, id).get();
        log.info("删除结果：{}", response.toString());
    }

    /**
     * 按照条件删除
     *
     * @param index 索引
     * @param type  类型
     * @param param 参数
     */
    public void deleteDocumentByQuery(String index, String type, UserSearchParam param) {
        DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE.newRequestBuilder(client);

        builder.source().setIndices(index).setTypes(type).setQuery(convertParam(param));
        BulkByScrollResponse response = builder.get();
        log.info("删除结果: {}", response.toString());
    }

    /**
     * 转换参数
     *
     * @param param 参数
     * @return 参数构造器
     */
    private BoolQueryBuilder convertParam(UserSearchParam param) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        if (StringUtils.hasText(param.getUserName())) {
            boolQueryBuilder.must(QueryBuilders.termQuery("userName", param.getUserName()));
        }
        if (param.getAge() != null) {
            boolQueryBuilder.must(QueryBuilders.rangeQuery("age").gt(param.getAge()));
        }
        if (StringUtils.hasText(param.getDescription())) {
            boolQueryBuilder.must(QueryBuilders.matchQuery("description", param.getDescription()));
        }
        if (StringUtils.hasText(param.getRoleName())) {
            boolQueryBuilder.must(QueryBuilders.nestedQuery("roles", QueryBuilders.termQuery("roles.name", param.getRoleName()), ScoreMode.None));
        }
        return boolQueryBuilder;
    }

    /**
     * 按照ID更新文档
     *
     * @param indices 指标
     * @param type    类型
     * @param id      ID
     * @param object  修改对象
     */
    public void updateDocument(String indices, String type, String id, Object object) {
        UpdateResponse response = client.prepareUpdate(indices, type, id).setDoc(JSON.toJSONString(object), XContentType.JSON).get();
        log.info("更新结果: {}", response.toString());
    }

    /**
     * 按条件更新
     *
     * @param indices 指标
     * @param type    类型
     * @param param   参数
     */
    public void updateDocumentByQuery(String indices, String type, UserSearchParam param) {
        UpdateByQueryRequestBuilder builder = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
        builder.source().setIndices(indices).setTypes(type).setQuery(convertParam(param));
    }

    /**
     * 按照条件查询
     *
     * @param indices 指标
     * @param type    类型
     * @param param   参数
     * @param clazz   字节码文件
     * @param <T>     泛型
     * @return 查询实体集合
     */
    public <T> List<T> queryDocumentByParam(String indices, String type, UserSearchParam param, Class<T> clazz) {
        SearchRequestBuilder builder = buildRequest(indices, type);
        assert builder != null;
        builder.addSort("birthday", SortOrder.ASC);
        builder.setQuery(convertParam(param));
        builder.setFrom(0).setSize(10);
        SearchResponse response = builder.get();
        return convertResponse(response, clazz);
    }

    /**
     * 通用的装换返回结果
     *
     * @param response 相应结果
     * @param clazz    结果字节码
     * @param <T>      结果类型
     * @return 结果集
     */
    private <T> List<T> convertResponse(SearchResponse response, Class<T> clazz) {
        List<T> list = new ArrayList<>();
        if (response != null && response.getHits() != null) {
            String result;
            Field idField = ReflectionUtils.findField(clazz, "id");
            if (idField != null) {
                ReflectionUtils.makeAccessible(idField);
            }
            T e = null;
            for (SearchHit hit : response.getHits()) {
                result = hit.getSourceAsString();

                if (StringUtils.hasText(result)) {
                    e = JSON.parseObject(result, clazz);
                }
                if (e != null) {
                    if (idField != null) {
                        ReflectionUtils.setField(idField, e, hit.getId());
                    }
                    list.add(e);
                }
            }
        }
        return list;
    }

    private SearchRequestBuilder buildRequest(String indices, String type) {
        return client.prepareSearch(indices).setTypes(type);
    }

    /**
     * 不存在就创建索引
     *
     * @param index 索引
     * @param type  类型
     * @return 是否存在
     */
    public boolean createIndexIfNotExist(String index, String type) {
        IndicesAdminClient adminClient = client.admin().indices();
        IndicesExistsRequest request = new IndicesExistsRequest(index);
        IndicesExistsResponse response = adminClient.exists(request).actionGet();
        if (!response.isExists()) {
            return createIndex(index, type);
        }
        return true;
    }

    /**
     * 创建索引
     *
     * @param index 索引
     * @param type  类型
     * @return 是否存在
     */
    public boolean createIndex(String index, String type) {
        XContentBuilder mappingBuilder;
        try {
            mappingBuilder = this.getMapping(type);
        } catch (Exception e) {
            log.error(String.format("创建Mapping 异常;index:%s type:%s;", index, type), e);
            return false;
        }
        Settings settings = Settings.builder()
                .put("index.number_of_shards", 2)
                .put("index.number_of_replicas", 1)
                .put("index.refresh_interval", "5s")
                .build();
        IndicesAdminClient adminClient = client.admin().indices();
        CreateIndexRequestBuilder builder = adminClient.prepareCreate(index);
        builder.setSettings(settings);
        CreateIndexResponse response = builder.addMapping(type, mappingBuilder).get();
        log.info("创建索引: {} 类型: {} 是否成功: {}", index, type, response.isAcknowledged());
        return response.isAcknowledged();
    }

    /**
     * 创建索引的Mapping信息  注意声明的roles为nested类型
     *
     * @param type 类型
     * @return 索引Mapping信息
     * @throws Exception 异常信息
     */
    public XContentBuilder getMapping(String type) throws Exception {
        return XContentFactory.jsonBuilder().startObject().startObject(type)
                .startObject("_all").field("enabled", false).endObject()
                .startObject("properties")
                .startObject("userName").field("type", "keyword").endObject()
                .startObject("age").field("type", "integer").endObject()
                .startObject("birthday").field("type", "date").endObject()
                .startObject("description").field("type", "text").field("analyzer", "ik_smart").endObject()
                .startObject("roles").field("type", "nested")
                .startObject("properties")
                .startObject("createTime").field("type", "date").endObject()
                .startObject("name").field("type", "keyword").endObject()
                .startObject("description").field("type", "text").field("analyzer", "ik_smart").endObject()
                .endObject()
                .endObject()
                .endObject().endObject().endObject();
    }
}
